package com.gmall.realtime.common.util;

import com.gmall.realtime.common.constant.Constant;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import com.ververica.cdc.debezium.StringDebeziumDeserializationSchema;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.io.IOException;
import java.lang.reflect.Type;
import java.nio.charset.StandardCharsets;
import java.util.Properties;

public class FlinkSourceUtil {

    /**
     * @param topic   主题
     * @param groupId 消费者组ID
     * @return
     */
    public static KafkaSource<String> getKafkaSource(String topic, String groupId) {
        return KafkaSource.<String>builder()
                // 设置Kafka服务器地址
                .setBootstrapServers(Constant.KAFKA_BROKERS)
                // 设置要订阅的主题
                .setTopics(topic)
                // 设置消费者组ID
                .setGroupId(groupId)
                // 设置初始偏移量为最早的消息
                .setStartingOffsets(OffsetsInitializer.earliest())
                // 设置反序列化器，只处理值部分
                .setValueOnlyDeserializer(new DeserializationSchema<String>() {
                    @Override
                    public String deserialize(byte[] bytes) throws IOException {
                        // 如果字节数组不为空，则将其转换为UTF-8编码的字符串
                        if (bytes != null) {
                            return new String(bytes, StandardCharsets.UTF_8);
                        }
                        // 如果字节数组为空，则返回null
                        return null;
                    }

                    @Override
                    public boolean isEndOfStream(String s) {
                        // 表示流没有结束
                        return false;
                    }

                    @Override
                    public TypeInformation<String> getProducedType() {
                        // 返回字符串类型的类型信息
                        return Types.STRING;
                    }
                })
                // 构建KafkaSource对象
                .build();
    }


    /**
     * 读取CDC
     *
     * @param processDataBase
     * @param processDimTableName
     * @return
     */
    public static MySqlSource<String> getMysqlSource(String processDataBase, String processDimTableName) {
        // 创建Properties对象
        Properties props = new Properties();
        // 设置是否使用SSL连接
        props.setProperty("useSSL", "false");
        // 设置是否允许获取公钥
        props.setProperty("allowPublicKeyRetrieval", "true");

        // 创建MySqlSource对象
        return MySqlSource.<String>builder()
                // 设置jdbc属性
                .jdbcProperties(props)
                // 设置主机名
                .hostname(Constant.MYSQL_HOST)
                // 设置端口号
                .port(Constant.MYSQL_PORT)
                // 设置数据库列表，监控inventory数据库下的所有表
                .databaseList(Constant.PROCESS_DATABASE) // monitor all tables under inventory database
                // 设置用户名
                .username(Constant.DORIS_USERNAME)
                // 设置密码
                .password(Constant.MYSQL_PASSWORD)
                // 设置表列表，格式为数据库名.表名
                .tableList(processDataBase + "." + processDimTableName)
                // 设置反序列化器，将SourceRecord转换为String
                .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to String
                // 设置启动选项，初始启动
                .startupOptions(StartupOptions.initial())
                // 构建MySqlSource对象
                .build();

    }

}
